package com.mq.mqserver;

import com.mq.common.*;
import com.mq.mqserver.core.BasicProperties;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 交换机 服务端
 * 当前程序只考虑一个虚拟主机
 */
public class BrokerServer {
    //调用相关数据
    private VirtualHost virtualHost = new VirtualHost("default");
    //服务器⾃⾝的 socket
    private ServerSocket serverSocket = null;
    //引入线程池,处理多个客户端的请求
    private ExecutorService executorService = null;
    //引入一个哈希表,存储所有的会话对象
    //key: channelId, val:socket对象
    private ConcurrentHashMap<String, Socket> sessions = new ConcurrentHashMap<>();
    //引入一个布尔变量,表示当前服务器是否要停止,
    //要对所有线程是立即可见的,用volatile修饰
    private volatile boolean runnable = true;

    public BrokerServer(int port) throws IOException {
        serverSocket = new ServerSocket(9090);
    }

    //启动服务
    public void start() throws IOException {
        System.out.println("[BrokerServer] 启动服务");
        executorService = Executors.newCachedThreadPool();
        try {
            while (runnable) {
                Socket clientSocket = serverSocket.accept();
                executorService.submit(() -> {
                    processConnection(clientSocket);
                });
            }
        } catch (SocketException e) {
            //正常结束
            System.out.println("[BrokerServer] 服务器停止运行!");
        }
    }

    //停止服务器
    public void stop() throws IOException {
        runnable = false;
        executorService.shutdown();
        serverSocket.close();
    }

    //处理一个客户端的连接
    //一个个连接可能有多次的请求和相应
    //要读取数据,处理数据,然后将结果返回给客户端
    private void processConnection(Socket clientSocket) {
        try (InputStream inputStream = clientSocket.getInputStream();
             OutputStream outputStream = clientSocket.getOutputStream()) {
            //  这里需要按照特定格式进行读取和解析数据
            try (DataInputStream dataInputStream = new DataInputStream(inputStream);
                 DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                while (true) {
                    //1.读取请求
                    Request request = readRequest(dataInputStream);
                    //2.根据请求计算相应
                    Response response = process(request, clientSocket);
                    //3.将结果返回给客户端
                    writeResponse(response, dataOutputStream);
                }
            } catch (EOFException | SocketException e) {
                //当出现这两种异常时,是正常的异常,是请求读取结束了,读到了空字符串抛出的异常,
                // 正常结束循环就可以了
                System.out.println("[BrokerServer] connection 连接关闭 ,客户端地址: " + clientSocket.getInetAddress().toString()
                        + " : " + clientSocket.getPort());
            } catch (ClassNotFoundException e) {
                throw new RuntimeException(e);
            } catch (MqException e) {
                throw new RuntimeException(e);
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            //关闭资源
            try {
                //当前连接处理完之后,需要关闭Socket
                clientSocket.close();
                //把当前socket对应的所有channel也删除了
                clearCloseSessions(clientSocket);
            } catch (IOException e) {
                e.printStackTrace();
            }
            //删除sessions中客户端和服务器建立的连接
        }
    }

    //  读取客户端请求
    private Request readRequest(DataInputStream dataInputStream) throws IOException {
        //读取数据
        int type = dataInputStream.readInt();
        int length = dataInputStream.readInt();
        byte[] payload = new byte[length];
        int n = dataInputStream.read(payload);
        if (n != length) {
            throw new IOException("[BrokerServer] 读取请求格式不正确");
        }
        //构造请求
        Request request = new Request();
        request.setType(type);
        request.setLength(length);
        request.setPayload(payload);
        return request;
    }

    /**
     * type 表⽰请求响应不同的功能. 取值如下
     * 0x1  创建 channel
     * • 0x2  关闭 channel
     * • 0x3  创建 exchange
     * • 0x4  销毁 exchange
     * • 0x5  创建 queue
     * • 0x6  销毁 queue
     * • 0x7  创建 binding
     * • 0x8  销毁 binding
     * • 0x9  发送 message
     * • 0xa  订阅 message
     * • 0xb  返回 ack
     * • 0xc  服务器给客⼾端推送的消息. (被订阅的消息) 响应独有的
     */
    // 处理请求,计算响应并返回
    private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {
        //1,先对请求进行解析,将payload的字节数组转换成对象
        BasicArgs basicArgs = (BasicArgs) BinaryTool.fromBytes(request.getPayload());
        //2.根据请求的类型,和之前的确定 做出响应
        int type = request.getType();
        boolean ok = true;
        if (type == 0x1) {
            //创建channel
            sessions.put(basicArgs.getChannelId(), clientSocket);
            System.out.println("[BrokerServer]  创建channel完成 channelId:" + basicArgs.getChannelId());
        } else if (type == 0x2) {
            //删除channel
            sessions.remove(basicArgs.getChannelId());
            System.out.println("[BrokerServer]  删除channel完成 channelId:" + basicArgs.getChannelId());
        } else if (type == 0x3) {
            //创建交换机
            ExchangeDeclareArgs exchangeDeclareArgs = (ExchangeDeclareArgs) basicArgs;
            ok = virtualHost.exchangeDeclare(exchangeDeclareArgs.getExchangeName(), exchangeDeclareArgs.getType(),
                    exchangeDeclareArgs.isDurable(), exchangeDeclareArgs.isAutoDelete(), exchangeDeclareArgs.getArgs());
            if(ok)  System.out.println("[BrokerServer]  创建交换机完成 exchangeName:" + exchangeDeclareArgs.getExchangeName());
            else
                System.out.println("[BrokerServer]  创建交换机失败 exchangeName:" + exchangeDeclareArgs.getExchangeName());
        } else if (type == 0x4) {
            //删除交换机
            ExchangeDeleteArgs exchangeDeleteArgs = (ExchangeDeleteArgs) basicArgs;
            ok = virtualHost.exchangeDelete(exchangeDeleteArgs.getExchangeName());
            if(ok) System.out.println("[BrokerServer]  删除交换机完成 exchangeName:" + exchangeDeleteArgs.getExchangeName());
            else System.out.println("[BrokerServer]  删除交换机失败 exchangeName:" + exchangeDeleteArgs.getExchangeName());
        } else if (type == 0x5) {
            //创建队列
            QueueDeclareArgs queueDeclareArgs = (QueueDeclareArgs) basicArgs;
            ok = virtualHost.queueDeclare(queueDeclareArgs.getQueueName(), queueDeclareArgs.isDurable(),
                    queueDeclareArgs.isAutoDelete(), queueDeclareArgs.isExclusive(), queueDeclareArgs.getArgs());
            if(ok) System.out.println("[BrokerServer]  创建队列完成 queueName:" + queueDeclareArgs.getQueueName());
            else System.out.println("[BrokerServer]  创建队列失败 queueName:" + queueDeclareArgs.getQueueName());
        } else if (type == 0x6) {
            //删除队列
            QueueDeleteArgs queueDeleteArgs = (QueueDeleteArgs) basicArgs;
            ok = virtualHost.queueDelete(queueDeleteArgs.getQueueName());
            if(ok) System.out.println("[BrokerServer]  删除队列完成 queueName:" + queueDeleteArgs.getQueueName());
            else System.out.println("[BrokerServer]  删除队列失败 queueName:" + queueDeleteArgs.getQueueName());
        } else if (type == 0x7) {
            //创建绑定
            BindingDeclareArgs bindingDeclareArgs = (BindingDeclareArgs) basicArgs;
            ok = virtualHost.bindingDeclare(((BindingDeclareArgs) basicArgs).getExchangeName(),
                    bindingDeclareArgs.getQueueName(), bindingDeclareArgs.getBindingKey());
            if(ok) System.out.println("[BrokerServer] 创建绑定成功 exchangeName:" + bindingDeclareArgs.getExchangeName()
                    + " ,queueName:" + bindingDeclareArgs.getQueueName() + " ,bindingKey:" + bindingDeclareArgs.getBindingKey());
            else System.out.println("[BrokerServer] 创建绑定失败 exchangeName:" + bindingDeclareArgs.getExchangeName()
                    + " ,queueName:" + bindingDeclareArgs.getQueueName() + " ,bindingKey:" + bindingDeclareArgs.getBindingKey());
        } else if (type == 0x8) {
            //删除绑定
            BindingDeleteArgs bindingDeleteArgs = (BindingDeleteArgs) basicArgs;
            ok = virtualHost.bindingDelete(bindingDeleteArgs.getExchangeName(), bindingDeleteArgs.getQueueName());
            if(ok) System.out.println("[BrokerServer] 删除绑定成功 exchangeName:" + bindingDeleteArgs.getExchangeName());
            else System.out.println("[BrokerServer] 删除绑定失败 exchangeName:" + bindingDeleteArgs.getExchangeName());
        } else if (type == 0x9) {
//            System.out.println("0x9");
            //发送消息
            BasicPublishArgs basicPublishArgs = (BasicPublishArgs) basicArgs;
            ok = virtualHost.basicPublish(basicPublishArgs.getExchangeName(), basicPublishArgs.getRoutingKey(),
                    basicPublishArgs.getBasicProperties(), basicPublishArgs.getBody());
            if(ok) System.out.println("[BrokerServe] 发送消息成功 messageId:"+basicPublishArgs.getBasicProperties().getMessageId());
            else System.out.println("[BrokerServer] 发送消息失败");
        } else if (type == 0xa) {
            //订阅消息
            BasicConsumerArgs basicConsumerArgs = (BasicConsumerArgs) basicArgs;
            ok = virtualHost.basicConsumer(basicConsumerArgs.getConsumerTag(), basicConsumerArgs.getQueueName(),
                    basicConsumerArgs.isAutoAck(), new Consumer() {
                        //这个回调函数要做的工作, 就是把服务器收到的消息可以直接推送回对应的消费者客户端
                        //此处 consumerTag 其实是 channelId. 根据 channelId 去 sessions 中查询,
                        //  就可以得到对应的socket 对象了, 从而可以往里面发送数据了
                        @Override
                        public void handlerDeliver(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {
                            //1.获取消费者客户端
                            Socket clientSocket = sessions.get(consumerTag);
                            if (clientSocket == null || clientSocket.isClosed()) {
                                throw new MqException("[BrokerServer] 订阅消息的消费者客户端已经不存在了");
                            }
                            //2.构造响应数据
                            SubScribeReturns subScribeReturns = new SubScribeReturns();
                            subScribeReturns.setConsumerTag(consumerTag);
                            subScribeReturns.setBasicProperties(basicProperties);
                            subScribeReturns.setBody(body);
                            //填写父类属性
                            //这个rid是用来确认每一次请求和确认消息的对应关系的,而这里的确认消息可能会有多次
                            //而不是一一对应的关系,只有响应,没有请求,rid就没有意义了,就先不写了
                            subScribeReturns.setRid("");
                            //consumerTag就是channelId
                            subScribeReturns.setChannelId(consumerTag);
                            subScribeReturns.setOk(true);
                            //3.返回给消费者客户端响应
                            //要通过网络传输,先将数据内容构造成二进制字节数组,再打包成响应对象
                            byte[] payload = BinaryTool.toByte(subScribeReturns);
                            Response response = new Response();
                            response.setType(0xc);
                            response.setLength(payload.length);
                            response.setPayload(payload);
                            //返回给客户端
                            DataOutputStream dataOutputStream = new DataOutputStream(clientSocket.getOutputStream());
                            //通过这个方法,返回响应给消费者 客户端
                            writeResponse(response, dataOutputStream);
                        }
                    });
        } else if (type == 0xb) {
            //返回Ack
            BasicAckArgs basicAckArgs = (BasicAckArgs) basicArgs;
            ok = virtualHost.basicAck(basicAckArgs.getQueueName(), basicAckArgs.getMessageId());
            if(ok) System.out.println("[BrokerServer] 返回Ack成功 queueName:"+basicAckArgs.getQueueName());
            else System.out.println("[BrokerServer] 返回Ack失败 queueName:"+basicAckArgs.getQueueName());
        } else {
            //类型异常
            throw new MqException("[BrokerServer] 未知type type:" + type);
        }
        //3.构造响应对象
        BasicReturns basicReturns = new BasicReturns();
        basicReturns.setRid(basicArgs.getRid());
        basicReturns.setChannelId(basicArgs.getChannelId());
        basicReturns.setOk(ok);
        byte[] payload = BinaryTool.toByte(basicReturns);

        Response response = new Response();
        response.setType(request.getType());
        response.setLength(payload.length);
        response.setPayload(payload);
        System.out.println("[BrokerServer] Response: rid=" + basicReturns.getRid() + ", channelId=" + basicReturns.getChannelId()
                + ", type=" + response.getType() + ", length=" + response.getLength());
        return response;
    }

    //将响应结果返回给客户端
    private void writeResponse(Response response, DataOutputStream dataOutputStream) throws IOException {
        dataOutputStream.writeInt(response.getType());
        dataOutputStream.writeInt(response.getLength());
        dataOutputStream.write(response.getPayload());
        //切记要刷新缓冲区,将全部数据从内存中放到文件中
        dataOutputStream.flush();
    }

    // 处理完请求,并返回响应后,删除客户端和服务器建立的连接关系
    //一个客户端可能建立多条连接
// 把该被关闭的 socket 对应的键值对, 统统删掉.
    private void clearCloseSessions(Socket clientSocket) {
        //存储要删除的Session键值对
        List<String> toRemoveChannelId = new ArrayList<>();
        for (Map.Entry<String, Socket> entry : sessions.entrySet()) {
            if (entry.getValue() == clientSocket) {
                //要将其删除,因为在遍历,不能直接从集合中删除,先保存起来
                //切记,集合不能一遍遍历,一遍删除
                toRemoveChannelId.add(entry.getKey());
            }
        }
        //然后再遍历删除
        for (String channelId : toRemoveChannelId) {
            sessions.remove(channelId);
        }
        System.out.println("[BrokerServer] 清理 session 完成! 被清理的 channelId=" + toRemoveChannelId);
    }
}
